package com.knight.hadoop.day09.comm.friend;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


public class QueryCommonFriendSecond {
	
	
	static class QueryCommonFriendSecondMapper extends Mapper<LongWritable, Text, Text, Text>{
		
		
		Text t1 = new Text();
		Text t2 = new Text();
		
		@Override
		protected void map(LongWritable key, Text value,Context context)
				throws IOException, InterruptedException {
			//A	I K C B G F H O D
			
			String line = value.toString();
			String[] friendPersons = line.split("\t");
			String persons = friendPersons[1];
			String friend = friendPersons[0];
			
			List<String> personList = Arrays.asList(persons.split(" "));
			
			Collections.sort(personList);
			
			for(int i=0;i<personList.size();i++){
				for(int j=i+1;j<personList.size();j++){
					t1.set(personList.get(i)+personList.get(j));
					t2.set(friend);
					context.write(t1, t2);
				}
			}
		}
	}
	
	static class QueryCommonFriendSecondReducer extends Reducer<Text, Text, Text, NullWritable>{
		
		Text t = new Text();
		
		@Override
		protected void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			List<String> friends = new ArrayList<>();
			
			for(Text t: values){
				friends.add(t.toString());
			}
			
			if(friends.size()>0){
				Collections.sort(friends);
				StringBuilder sb = new StringBuilder();
				for(int i=0;i<friends.size();i++){
					if(i==(friends.size()-1)){
						sb.append(friends.get(i));
					}else{
						sb.append(friends.get(i)+",");
					}
				}
				t.set(key+":"+sb.toString());
				context.write(t, NullWritable.get());
			}
		}
	}
	
	public static void main(String[] args) throws Exception {
		
		Configuration configuration = new Configuration();

		Job job = Job.getInstance(configuration);

		job.setJarByClass(QueryCommonFriendSecond.class);
		
		// 设置mapper任务执行类
		job.setMapperClass(QueryCommonFriendSecondMapper.class);
		// 设置reducer任务执行类
		job.setReducerClass(QueryCommonFriendSecondReducer.class);

		// 指定mapper输出数据的kv类型
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);

		// 指定最终输出数据的kv类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NullWritable.class);

		// 指定输入参数的目录
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		// 指定输出参数的目录
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		// 将job配置的参数以及job所用的java类所在的jar包，提交到yarn去运行
		/* job.submit(); */
		// 但是我们一般用这个，因为可以等待运行结果返回，查看运行流程
		boolean res = job.waitForCompletion(true);
		System.exit(res ? 0 : 1);
	}
}
